package com.zhao.utils;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueExcludeFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * Description: <br/>
 * Copyright (c) ，2020 ， 赵 <br/>
 * This program is protected by copyright laws. <br/>
 * Date： 2020/9/19 18:13
 *
 * @author 柒柒
 * @version : 1.0
 */

public class HbaseUtils {
    /**
     * 获取Configuration对象
     */
    private static Connection connection = null;

    static{

        try {
            Configuration conf = HBaseConfiguration.create();
            //使用HBaseConfiguration的单例方法实例化
            conf.set("hbase.zookeeper.quorum", "node01,node02,node03");
            conf.set("hbase.zookeeper.property.clientPort", "2181");
            connection= ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    public static void main(String[] args) throws IOException {
        //判断表是否存在
        //tableExists("student");

        //删除表
        //dropTable("student");

        //创建表
        //create 'student','info'
        //creatTable("student","info");

        //向表中插入数据
//        put 'student','1001','info:sex','male'
//        put 'student','1001','info:age','18'
//        put 'student','1002','info:name','buwenbuhuo'
//        put 'student','1002','info:sex','female'
//        put 'student','1002','info:age','20'
        //putCell("student","1005","info", "name","lixuefang");
        //putCell("student","1005","info", "gender","girl");
        //putCell("student","1005","info", "age","18");

        //删除某一行
        //deleteRow("student","1005");

        //删除某个列族
        //deleteFamily("student","1005","info");

        //删除某一行
        //deleteCell("student","1005","info","name");

        //删除最新版本
        //deleteCells("student","1005","info", "name");

        //查找一行数据
        //getRow("student","1005");

        //查找指定范围内的数据
        //getRows("student","1001","1005");
        //getRows("student","1001","1005!");

        //过滤单个
        //getRowsByColumn("student","info","age","18");

        //过滤多个
        HashMap<String,String>map = new HashMap<String, String>();
        map.put("age","18");
        map.put("gender","girl");
        getRowsByColumns("student","info",map);
    }

    /**
     * 判断表是否存在
     * @param tableName
     * @return
     * @throws IOException
     */
    public static boolean tableExists(String tableName) throws IOException{

        Admin admin = connection.getAdmin();

        try {
            return admin.tableExists(TableName.valueOf(tableName));
        }finally {
            admin.close();
        }

    }


    /**
     *  创建ddl
     * @param tableName
     * @param families
     * @throws IOException
     */
    public static void creatTable(String tableName,String... families)throws IOException {

        Admin admin = connection.getAdmin();

        try {
            if (admin.tableExists(TableName.valueOf(tableName))){
                System.out.println("表:"+tableName+" 已经存在");
                return;
            }

            HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(tableName));

            for (String family : families){
                HColumnDescriptor familyDesc = new HColumnDescriptor(family);
                desc.addFamily(familyDesc);
            }

            admin.createTable(desc);
        }finally {
            admin.close();
        }

    }


    /**
     * 删除表
     * @param tableName
     * @throws IOException
     */
    public static void dropTable(String tableName) throws IOException{

        Admin admin = connection.getAdmin();

        try{
            if(!admin.tableExists(TableName.valueOf(tableName))){
                System.out.println("表" + tableName + "删除成功！");
            }else{
                System.out.println("表" + tableName + "不存在！");
            }
            admin.disableTable(TableName.valueOf(tableName));

            admin.deleteTable(TableName.valueOf(tableName));
        }finally {
            admin.close();
        }

    }


    /**
     * 往表中插入数据
     * @param tableName
     * @param rowKey
     * @param family
     * @param column
     * @param value
     * @throws IOException
     */
    public static void putCell(String tableName,String rowKey,String family,String column,String value) throws IOException {

        if(!tableExists(tableName)){
            return;
        }
        Table table = connection.getTable(TableName.valueOf(tableName));
        try{


            Put put = new Put(Bytes.toBytes(rowKey));

            put.addColumn(Bytes.toBytes(family),Bytes.toBytes(column),Bytes.toBytes(value));

            table.put(put);
        }finally {
            table.close();
        }

    }


    /**
     *  1. 删除某一行
     * @param tableName
     * @param rowKey
     * @throws IOException
     */
    public static void deleteRow(String tableName,String rowKey) throws IOException {

        Table table = connection.getTable(TableName.valueOf(tableName));

        Delete delete = new Delete(Bytes.toBytes(rowKey));

        table.delete(delete);

        table.close();
    }

    /**
     *  2. 删除整个列组
     * @param tableName
     * @param rowKey
     * @param family
     * @throws IOException
     */
    public static void deleteFamily(String tableName,String rowKey,String family) throws IOException {

        Table table = connection.getTable(TableName.valueOf(tableName));

        Delete delete = new Delete(Bytes.toBytes(rowKey));

        delete.addFamily(Bytes.toBytes(family));

        table.delete(delete);

        table.close();
    }


    /**
     * 删除所有版本
     * @param tableName
     * @param rowKey
     * @param family
     * @param column
     * @throws IOException
     */
    public static void deleteCell(String tableName,String rowKey,String family,String column) throws IOException {

        Table table = connection.getTable(TableName.valueOf(tableName));

        Delete delete = new Delete(Bytes.toBytes(rowKey));

        delete.addColumn(Bytes.toBytes(family),Bytes.toBytes(column));

        table.delete(delete);

        table.close();
    }


    /**
     * 删除最新的版本
     * @param tableName
     * @param rowKey
     * @param family
     * @param column
     * @throws IOException
     */
    public static void deleteCells(String tableName,String rowKey,String family,String column) throws IOException {

        Table table = connection.getTable(TableName.valueOf(tableName));

        Delete delete = new Delete(Bytes.toBytes(rowKey));

        delete.addColumns(Bytes.toBytes(family),Bytes.toBytes(column));

        table.delete(delete);

        table.close();
    }

    /**
     * 查找一行数据
     * @param tableName
     * @param rowKey
     * @throws IOException
     */
    public static void getRow(String tableName,String rowKey) throws IOException {

        Table table = connection.getTable(TableName.valueOf(tableName));

        Get get = new Get(Bytes.toBytes(rowKey));

        Result result = table.get(get);

        Cell[] cells = result.rawCells();

        for (Cell cell : cells){

            byte[] columnBytes = CellUtil.cloneQualifier(cell);

            String columnStr = Bytes.toString(columnBytes);

            byte[] valueBytes = CellUtil.cloneValue(cell);

            String valueStr = Bytes.toString(valueBytes);

            System.out.println(columnStr + ":" + valueStr);
        }
        table.close();
    }

    /**
     * 查找执行范围内的数据
     * @param tableName
     * @param startRow
     * @param stopRow
     * @throws IOException
     */
    public static void getRows(String tableName,String startRow,String stopRow) throws IOException {

        Table table = connection.getTable(TableName.valueOf(tableName));

        Scan scan = new Scan(Bytes.toBytes(startRow), Bytes.toBytes(stopRow));

        ResultScanner scanner = table.getScanner(scan);

        for (Result result : scanner) {

            for (Cell cell : result.rawCells()) {

                String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
                String column = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));

                System.out.println(rowKey + "-" + column + ":" + value);
            }
        }

        scanner.close();
        table.close();

    }

    /**
     * 过滤单条数据
     * @param tableName
     * @param family
     * @param column
     * @param value
     * @throws IOException
     */
    public static  void getRowsByColumn(String tableName,String family,String column, String value) throws IOException {

        Table table = connection.getTable(TableName.valueOf(tableName));

        Scan scan = new Scan();

        SingleColumnValueExcludeFilter filter = new SingleColumnValueExcludeFilter(Bytes.toBytes(family), Bytes.toBytes(column), CompareFilter.CompareOp.EQUAL, Bytes.toBytes(value));
        filter.setFilterIfMissing(true);  // 过滤器
        scan.setFilter(filter);

        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {

            for (Cell cell : result.rawCells()) {

                String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
                String columnStr = Bytes.toString(CellUtil.cloneQualifier(cell));
                String valueStr = Bytes.toString(CellUtil.cloneValue(cell));

                System.out.println(rowKey + "-" + columnStr + ":" + valueStr);
            }
        }

        scanner.close();
        table.close();

    }

    /**
     * 过滤多条数据
     * @param tableName
     * @param family
     * @param map
     * @throws IOException
     */
    public static  void getRowsByColumns(String tableName, String family, Map<String,String> map) throws IOException {

        Table table = connection.getTable(TableName.valueOf(tableName));

        Scan scan = new Scan();

        SingleColumnValueExcludeFilter filter1 = new SingleColumnValueExcludeFilter(Bytes.toBytes(family), Bytes.toBytes(map.keySet().toArray()[0].toString()), CompareFilter.CompareOp.EQUAL, Bytes.toBytes(map.get(map.keySet().toArray()[0].toString())));
        filter1.setFilterIfMissing(true);

        SingleColumnValueExcludeFilter filter2 = new SingleColumnValueExcludeFilter(Bytes.toBytes(family), Bytes.toBytes(map.keySet().toArray()[1].toString()), CompareFilter.CompareOp.EQUAL, Bytes.toBytes(map.get(map.keySet().toArray()[1].toString())));
        filter2.setFilterIfMissing(true);

        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);

        filterList.addFilter(filter1);
        filterList.addFilter(filter2);

        scan.setFilter(filterList);

        // da yin
        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {

            for (Cell cell : result.rawCells()) {

                String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
                String columnStr = Bytes.toString(CellUtil.cloneQualifier(cell));
                String valueStr = Bytes.toString(CellUtil.cloneValue(cell));

                System.out.println(rowKey + "-" + columnStr + ":" + valueStr);
            }
        }

        scanner.close();
        table.close();

    }




}
